327a19abf7cc8e420a293214848dccc109b6012e,cdap-app-templates/cdap-etl/cdap-etl-batch/src/main/java/co/cask/cdap/templates/etl/batch/ETLBatchTemplate.java,ETLBatchTemplate,validateAdapter,#ETLStage#ETLStage#List#,140
Before Change
if (transformList.size() == 0) {
// No transforms. Check only source and sink.
if (!(TypeToken.of(batchSink.getKeyType()).isAssignableFrom(batchSource.getKeyType()) &&
TypeToken.of(batchSink.getValueType()).isAssignableFrom(batchSource.getValueType()))) {
throw new IllegalArgumentException(String.format("Source %s and Sink %s Types don't match",
source.getName(), sink.getName()));
}
} else {
// Check the first and last transform with source and sink.
ETLStage firstStage = Iterables.getFirst(transformList, null);
ETLStage lastStage = Iterables.getLast(transformList);
Transform firstTransform = Iterables.getFirst(transforms, null);
Transform lastTransform = Iterables.getLast(transforms);
if (!(TypeToken.of(firstTransform.getKeyInType()).isAssignableFrom(batchSource.getKeyType()) &&
TypeToken.of(firstTransform.getValueInType()).isAssignableFrom(batchSource.getValueType()))) {
throw new IllegalArgumentException(String.format("Source %s and Transform %s Types don't match",
source.getName(), firstStage.getName()));
}
After Change
Transform lastTransform = Iterables.getLast(transforms);
if (!(isAssignable(batchSource.getKeyType(), firstTransform.getKeyInType()) &&
(isAssignable(batchSource.getValueType(), firstTransform.getValueInType())))) {
throw new IllegalArgumentException(String.format("Source %s and Transform %s Types don't match",
source.getName(), firstStage.getName()));
}